package util;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;

/**
 * Hadoop operators: managing files on HDFS
 * @author tigerzhong
 *
 */
public class HadoopUtil {

    /**
     * Send file to HDFS
     * 
     * @param src
     * @param dst
     */
    public static void localFile2HDFS(String src, String dst) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    Path dstPath = new Path(dst);
	    fs.copyFromLocalFile(srcPath, dstPath);
	} catch (IOException e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }

    /**
     * Get file from HDFS
     * 
     * @param delSrc
     * @param src
     * @param dst
     */
    public static void HDFSFile2Local(boolean delSrc, String src, String dst) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    Path dstPath = new Path(dst);
	    fs.copyToLocalFile(delSrc, srcPath, dstPath);
	} catch (IOException e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }

    /**
     * Move files on the HDFS
     * 
     * @param delSrc
     * @param src
     * @param dst
     */
    public static void HDFSMove(boolean delSrc, String src, String dst) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    FileStatus fsta[] = fs.globStatus(srcPath);
	    for (FileStatus it : fsta) {
		Path singlePath = it.getPath();
		Path dstPath = new Path(dst);
		FileUtil.copyMerge(fs, singlePath, fs, dstPath, delSrc, conf,"");
	    }
	} catch (IOException e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }

    /**
     * Delete file from the HDFS
     * 
     * @param src
     */
    public static void HDFSDelete(String src) {
	FileSystem fs = null;
	Configuration conf = new Configuration();
	try {
	    fs = FileSystem.get(conf);
	    Path srcPath = new Path(src);
	    fs.delete(srcPath, true);
	} catch (IOException e) {
	    e.printStackTrace();
	} finally {
	    try {
		fs.close();
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }
}
